#include "app_mqtt.h"
#include "thirdparty/log/log.h"
#include <MQTTClient.h>
#include <stdlib.h>
#include <assert.h>

// mqtt 客户端
static MQTTClient client;
// mqtt 连接参数
static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
// mqtt 消息结构体
static MQTTClient_message pubmsg = MQTTClient_message_initializer;
// mqtt 令牌
static MQTTClient_deliveryToken token;
// 我们自定义的信息处理回调函数
static int (*recv_callback)(char *, int);

void app_mqtt_deliveredCallback(void *context, MQTTClient_deliveryToken dt)
{
    assert(context == NULL);
    log_trace("Message with token value %d delivery confirmed", dt);
}

int app_mqtt_messageArrivedCallback(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    assert(context == NULL);
    log_trace("Message arrived: topic %s, message %s", topicName, message->payload);

    // 调用回调函数 这个回调函数会将数据写入到设备缓存
    int result = recv_callback(message->payload, message->payloadlen);
    return result == 0 ? 1 : 0;
}
void app_mqtt_connectionLostCallback(void *context, char *cause)
{
    assert(context == NULL);
    log_warn("Connection lost because of %s", cause);
}

int app_mqtt_init(void)
{
    // 初始化客户端
    if (MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL) != MQTTCLIENT_SUCCESS)
    {
        log_error("Mqtt create fail");
        goto MQTT_CREATE_FAIL;
    }

    // 设置回调函数
    if (MQTTClient_setCallbacks(client, NULL, app_mqtt_connectionLostCallback, app_mqtt_messageArrivedCallback, app_mqtt_deliveredCallback) != MQTTCLIENT_SUCCESS)
    {
        log_error("Mqtt setCallbacks fail");
        goto MQTT_CONNECTION_FAIL;
    }

    // 连接
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if (MQTTClient_connect(client, &conn_opts) != MQTTCLIENT_SUCCESS)
    {
        log_error("Mqtt connection fail");
        goto MQTT_CONNECTION_FAIL;
    }

    // 订阅
    if (MQTTClient_subscribe(client, PULL_TOPIC, QOS) != MQTTCLIENT_SUCCESS)
    {
        log_error("Mqtt subscribe fail");
        goto MQTT_SUBSCRIBE_FAIL;
    }
    return 0;

MQTT_SUBSCRIBE_FAIL:
    MQTTClient_disconnect(client, TIMEOUT);
MQTT_CONNECTION_FAIL:
    MQTTClient_destroy(&client);
MQTT_CREATE_FAIL:
    return -1;
}

void app_mqtt_close(void)
{
    MQTTClient_disconnect(client, TIMEOUT);
    MQTTClient_destroy(&client);
}

int app_mqtt_send(char *payload, int payload_len)
{
    pubmsg.payload = payload;
    pubmsg.payloadlen = payload_len;
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    int result = MQTTClient_publishMessage(client, PUSH_TOPIC, &pubmsg, &token);
    if (result == MQTTCLIENT_SUCCESS)
    {
        log_trace("Message with delivery token %d delivered", token);
        return 0;
    }
    else
    {
        log_warn("Message with delivery token %d failed", token);
        return -1;
    }
}

void app_mqtt_registerRecvCallback(int (*callback)(char *payload, int payload_len))
{
    recv_callback = callback;
}
